/*
 * Copyright (C) 2017 Pingan, Inc. All Rights Reserved.
 */

package com.pingan.haofang.searchcloud.es.indices.impl;

import com.pingan.haofang.searchcloud.api.IndexDataPacket;
import com.pingan.haofang.searchcloud.api.IndexRowData;
import com.pingan.haofang.searchcloud.es.api.client.EsClientFactory;
import com.pingan.haofang.searchcloud.es.api.excption.DataPushException;
import com.pingan.haofang.searchcloud.es.constants.EsConstants;
import com.pingan.haofang.searchcloud.es.indices.IndicesDataService;
import org.apache.commons.collections.CollectionUtils;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.index.VersionType;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.metrics.max.Max;
import org.elasticsearch.search.aggregations.metrics.max.MaxAggregationBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.List;
import java.util.Map;

/**
 * IndicesDataServiceImpl
 *
 * @author zhangbi617
 * @date 2017-07-05
 */
@Service
public class IndicesDataServiceImpl implements IndicesDataService {

    private static final Logger LOGGER = LoggerFactory.getLogger(IndicesDataServiceImpl.class);

    @Autowired
    EsClientFactory esClientFactory;


    @Override
    public void create(String host, String indexName, String type, String id, String json) {
        esClientFactory.getByHost(host).prepareIndex(indexName, type, id)
                .setSource(json, XContentType.JSON).get();
    }

    @Override
    public String getData(String host, String indexName, String type, String id) {
        return esClientFactory.getByHost(host).prepareGet(indexName, type, id)
                .get().getSourceAsString();
    }

    @Override
    public void deleteData(String host, String indexName, String type, String id) {
        esClientFactory.getByHost(host).prepareDelete(indexName, type, id)
                .get();
    }

    @Override
    public void bulkUpdate(String host, String indexName, String type, IndexDataPacket indexDataPacket) throws IOException {
        TransportClient transportClient = esClientFactory.getByHost(host);

        // 1、更新数据
        List<IndexRowData> updateData = indexDataPacket.getUpdateData();
        if (CollectionUtils.isNotEmpty(updateData)) {
            BulkRequestBuilder bulkRequest = transportClient.prepareBulk();
            for (IndexRowData row : updateData) {
                // 拼接数据
                XContentBuilder jsonBuilder = XContentFactory.jsonBuilder().startObject();
                for (Map.Entry<String, Object> entry : row.entrySet()) {
                    jsonBuilder.field(entry.getKey(), entry.getValue());
                }
                jsonBuilder.endObject();
                bulkRequest.add(
                        transportClient.prepareIndex(indexName, type, row.getPkFieldValue().toString())
                                .setVersion(row.getVersion())
                                .setVersionType(VersionType.EXTERNAL)
                                .setSource(jsonBuilder));
            }
            BulkResponse bulkResponse = bulkRequest.get();
            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("bulk update took,{}", bulkResponse.getTook().getSeconds());
            }

            if (bulkResponse.hasFailures()) {
                String failureMessage = bulkResponse.buildFailureMessage();
                LOGGER.info("bulk failure:{}", failureMessage);
                throw new DataPushException(failureMessage);
            }
        }

        // 2. 删除数据
        List<IndexRowData> deleteData = indexDataPacket.getDeleteData();
        if (CollectionUtils.isNotEmpty(deleteData)) {
            BulkRequestBuilder bulkRequest = transportClient.prepareBulk();
            for (IndexRowData rowData : deleteData) {
                bulkRequest.add(transportClient.prepareDelete(indexName, type, rowData.getPkFieldValue().toString()));
            }
            BulkResponse bulkResponse = bulkRequest.get();

            if (LOGGER.isDebugEnabled()) {
                LOGGER.debug("bulk delete took,{}", bulkResponse.getTook().getSeconds());
            }
            if (bulkResponse.hasFailures()) {
                String failureMessage = bulkResponse.buildFailureMessage();
                LOGGER.info("bulk failure:{}", failureMessage);
                throw new DataPushException(failureMessage);
            }
        }

    }

    @Override
    public Long getMaxValue(String host, String indexName, String versionField) {
        final String maxVersion = "maxVersion";
        SearchRequestBuilder requestBuilder = esClientFactory.getByHost(host)
                .prepareSearch(EsConstants.INDEX_PREFIX + indexName);
        //.setTypes(type);
        MaxAggregationBuilder aggregationBuilder = AggregationBuilders.max(maxVersion).field(versionField);
        requestBuilder.setSize(1);
        requestBuilder.setFrom(0);
        SearchResponse execute = requestBuilder.addAggregation(aggregationBuilder).execute().actionGet();

        Max max = execute.getAggregations().get(maxVersion);
        return Double.valueOf(max.getValue()).longValue() < 0 ? 0 : Double.valueOf(max.getValue()).longValue();
    }

}
